/**
 * 抖音关注：程序员三丙
 * 知识星球：https://t.zsxq.com/j9b21
 */
package sanbing.example.dylike.application.rpc;

import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.Futures;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;
import sanbing.example.dylike.domain.SeckillGateway;
import sanbing.example.dylike.infrastructure.queue.QueueConsumer;
import sanbing.example.dylike.infrastructure.queue.QueueMsg;
import sanbing.example.dylike.infrastructure.queue.QueueProducer;
import sanbing.example.dylike.infrastructure.queue.provider.QueueFactory;
import sanbing.example.dylike.infrastructure.queue.rpc.DefaultQueueResponseTemplate;
import sanbing.example.dylike.infrastructure.queue.rpc.QueueResponseTemplate;
import sanbing.example.dylike.util.JacksonUtil;
import sanbing.example.dylike.util.MessagesStats;
import sanbing.example.dylike.util.SanbingExecutors;
import sanbing.example.dylike.util.StatsFactory;

import java.util.UUID;
import java.util.concurrent.ExecutorService;

import static sanbing.example.dylike.application.rpc.RpcMethod.SECKILL;


@Slf4j
@Service
public class RpcResponseQueueService {

    @Resource
    QueueFactory queueFactory;
    @Resource
    StatsFactory statsFactory;
    @Resource
    SeckillGateway seckillGateway;

    @Value("${queue.rpc.max_pending_requests:10000}")
    private int maxPendingRequests;
    @Value("${queue.rpc.max_requests_timeout:10000}")
    private long requestTimeout;
    @Value("${queue.rpc.request_poll_interval:25}")
    private int responsePollDuration;
    @Value("${queue.rpc.max_callback_threads:100}")
    private int maxCallbackThreads;

    private ExecutorService rpcCallbackExecutor;
    private QueueResponseTemplate queueResponseTemplate;


    @PostConstruct
    public void init() {
        this.rpcCallbackExecutor = SanbingExecutors.newWorkStealingPool(maxCallbackThreads, getClass());
        QueueProducer producer = queueFactory.createSeckillResponseProducer();
        QueueConsumer consumer = queueFactory.createSeckillRpcRequestConsumer();

        MessagesStats queueStats = statsFactory.createMessagesStats("queue-rpc");

        DefaultQueueResponseTemplate.DefaultQueueResponseTemplateBuilder builder = DefaultQueueResponseTemplate.builder();
        builder.requestTemplate(consumer);
        builder.responseTemplate(producer);
        builder.maxPendingRequests(maxPendingRequests);
        builder.requestTimeout(requestTimeout);
        builder.pollInterval(responsePollDuration);
        builder.executor(rpcCallbackExecutor);
        builder.stats(queueStats);
        queueResponseTemplate = builder.build();
    }

    @EventListener(ApplicationReadyEvent.class)
    @Order(10)
    public void onApplicationEvent(ApplicationReadyEvent event) {
        log.info("Received application ready event. Starting polling for events.");
        queueResponseTemplate.init(request -> {

            RpcRequest rpcRequest = JacksonUtil.OBJECT_MAPPER.readValue(request.getData(), RpcRequest.class);

            // todo 注解+改策略模式
            if (SECKILL == rpcRequest.getMethod()) {

                JsonNode params = rpcRequest.getParams();

                QueueMsg queueMsg = new QueueMsg(UUID.randomUUID(),
                        new RpcResponse(seckillGateway.write(UUID.fromString(params.get("skuId").asText()))).toBytes());

                return Futures.immediateFuture(queueMsg);
            }

            return Futures.immediateFailedFuture(new RuntimeException("unknown rpc request"));
        });
    }

    @PreDestroy
    public void destroy() {
        if (queueResponseTemplate != null) {
            queueResponseTemplate.stop();
        }
        if (rpcCallbackExecutor != null) {
            rpcCallbackExecutor.shutdownNow();
        }
    }

}